package com.lagou.client;

import com.lagou.handler.RpcEncoder;
import com.lagou.handler.UserClientHandler;
import com.lagou.request.RpcRequest;
import com.lagou.serializer.JSONSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 消费者
 */
public class RPCConsumer {

    ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();

    //1.创建一个线程池对象  -- 它要处理我们自定义事件
    private static ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    public HashMap<String, ChannelFuture> channelHashMap = new HashMap<>();

    public HashMap<String, Integer> key_time = new HashMap<>();

    public HashMap<String, ConsumerThread> key_thread = new HashMap<>();

    private static Bootstrap bootstrap;


    public Object createNewProxy(UserClientHandler handler, Class<?> serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    @Override
                    public Object invoke(Object o, Method method, Object[] objects) throws Throwable {

                        //2)给UserClientHandler 设置param参数
                        handler.setRpcRequest((RpcRequest) objects[0]);

                        //3).使用线程池,开启一个线程处理处理call() 写操作,并返回结果
                        Object result = executorService.submit(handler).get();

                        //4)return 结果
                        return result;
                    }
                });
    }

    //初始化netty客户端
    public void initClient() throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true);

    }

    public void start(String ip, int port) {
        try {
            UserClientHandler userClientHandler = new UserClientHandler();
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                    //设置编码
                    pipeline.addLast(new StringEncoder());
                    pipeline.addLast(new StringDecoder());
                    //添加自定义事件处理器
                    pipeline.addLast(userClientHandler);
                }
            });
            ChannelFuture future = bootstrap.connect(ip, port).sync();
            System.out.println("连接到服务器：" + ip + ":" + port);
            String key = ip + ":" + port;
            channelHashMap.put(key, future);
            ConsumerThread consumerThread = new ConsumerThread(userClientHandler, this);
            key_thread.put(key, consumerThread);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //开始发送
    public void startSend() {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                //选择一个响应最快的
                ConsumerThread thread = getThread();
                if (thread == null) {
                    return;
                }
                executorService.execute(thread);
            }
        };
        service.scheduleAtFixedRate(runnable, 10, 2, TimeUnit.SECONDS);
    }

    public ConsumerThread getThread() {
        int low = Integer.MAX_VALUE;
        String key = "";
        for (Map.Entry<String, Integer> entry : key_time.entrySet()) {
            if (entry.getValue() <= low) {
                key = entry.getKey();
                low = entry.getValue();
            }
        }
        System.out.println("==============================》》请求服务器:" + key);
        return key_thread.get(key);
    }


    public HashMap<String, ChannelFuture> getChannelHashMap() {
        return channelHashMap;
    }

    public void setChannelHashMap(HashMap<String, ChannelFuture> channelHashMap) {
        this.channelHashMap = channelHashMap;
    }

    public HashMap<String, Integer> getKey_time() {
        return key_time;
    }

    public void setKey_time(HashMap<String, Integer> key_time) {
        this.key_time = key_time;
    }

    public HashMap<String, ConsumerThread> getKey_thread() {
        return key_thread;
    }

    public void setKey_thread(HashMap<String, ConsumerThread> key_thread) {
        this.key_thread = key_thread;
    }
}
